package cn.doitedu.features;

import cn.doitedu.utils.Array2HbaseSerUDF;
import cn.doitedu.utils.Array2HbaseSerUDF2;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job02_同步请求特征日志到hbase {

    // 请求特征日志: 是广告引擎的后端日志
    // 它被flume采集到了kafka，topic: ad_request_log
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射kafka中的 "广告请求特征日志"
        tenv.executeSql(
                " create table ad_request_log_kafka(           \n "+
                        "   ad_tracking_id  string,            \n "+
                        " 	requestTime  bigint,               \n "+
                        " 	platform string,                   \n "+
                        " 	pageId string,                     \n "+
                        " 	locationType string,               \n "+
                        " 	mediaType string,                  \n "+
                        " 	adids array<double>,               \n "+
                        " 	ad_features array<array<double>>,  \n "+
                        " 	user_features array<double>,       \n "+
                        " 	device_features array<double>,     \n "+
                        " 	poster_features array<double>      \n "+
                        ") WITH (                                             \n" +
                        "    'connector' = 'kafka',                           \n" +
                        "    'topic' = 'ad_request_log',                      \n" +
                        "    'properties.bootstrap.servers' = 'doitedu:9092', \n" +
                        "    'properties.group.id' = 'doit44_g1',             \n" +
                        "    'scan.startup.mode' = 'latest-offset',           \n" +
                        "    'value.format' = 'json',                         \n" +
                        "    'value.fields-include' = 'EXCEPT_KEY'            \n" +
                        ")                                                    ");

        // 建表，映射hbase中的物理表  ad_request_log
        tenv.executeSql(
                        " create table ad_request_log_hbase(          \n "+
                        "   ad_tracking_id  string,                   \n "+
                        "   f ROW<                                    \n "+
                        " 	       requestTime  bigint,               \n "+
                        " 	       platform string,                   \n "+
                        " 	       pageId string,                     \n "+
                        " 	       locationType string,               \n "+
                        " 	       mediaType string,                  \n "+
                        " 	       adids string,               \n "+
                        " 	       ad_features string,         \n "+
                        " 	       user_features string,       \n "+
                        " 	       device_features string,     \n "+
                        " 	       poster_features string      \n "+
                        " 	>,                                        \n "+
                        " 	PRIMARY KEY(ad_tracking_id) not enforced           \n "+
                        " ) WITH (                                             \n "+
                        "  'connector' = 'hbase-2.2',                          \n "+
                        "  'table-name' = 'ad_request_log',                    \n "+
                        "  'zookeeper.quorum' = 'doitedu:2181'                 \n "+
                        " )                                                    \n ");




        // insert into ... select  from ....
        tenv.createTemporaryFunction("arr_ser", Array2HbaseSerUDF.class);
        tenv.createTemporaryFunction("arr_ser2", Array2HbaseSerUDF2.class);

        tenv.executeSql(
                " insert into ad_request_log_hbase                          "+
                        " with tmp as (                                             "+
                        " select                                                    "+
                        "       ad_tracking_id,                                   "+
                        " 	    requestTime,                                      "+
                        " 		platform,                                         "+
                        " 		pageId,                                           "+
                        " 		locationType,                                     "+
                        "       mediaType,                                        "+
                        " 		arr_ser(adids) as adids,                          "+
                        " 		arr_ser2(ad_features) as ad_features,             "+
                        " 		arr_ser(user_features) as user_features,          "+
                        "       arr_ser(device_features) as device_features,      "+
                        " 		arr_ser(poster_features) as poster_features       "+
                        " from ad_request_log_kafka                                 "+
                        " )                                                         "+
                        " 														    "+
                        " select                                                    "+
                        "    ad_tracking_id,                                        "+
                        "    row(requestTime,                                       "+
                        "    platform,                                              "+
                        "    pageId,                                                "+
                        "    locationType,                                          "+
                        "    mediaType,                                             "+
                        "    adids,                                                 "+
                        "    ad_features,                                           "+
                        "    user_features,                                         "+
                        "    device_features,                                       "+
                        "    poster_features) as f                                  "+
                        " from tmp                                                  "
        );


    }

}
